package re

import (
	"encoding/json"
	"fmt"
	"github.com/nsqio/go-nsq"
	"log/dingding"
	"log/es"
	"log/lg"
	"log/nacos"
	"log/shell"
	"regexp"
)

//正则消费任务结构体
type reTask struct {
	tail  chan string //日志通道
	gops  chan string
	re    *regexp.Regexp //正则表达式对象
	i     int            //携程并发数
	topic []*nacos.Conf
	addr  string
}

var reg *reTask

var js *shell.GoPs

// Init 初始化
func Init(re string, topic []*nacos.Conf, addr string) {
	reg = &reTask{
		tail:  make(chan string, 10000),
		gops:  make(chan string, 1000),
		re:    regexp.MustCompile(re),
		i:     10,
		topic: topic,
		addr:  addr,
	}
	//并发运行nsq消费
	reg.run()

}

//后台运行的
func (r *reTask) run() {

	for i, conf := range r.topic {
		//运行的nsq消费
		go r.StartConsumer(conf.Topic, i)
		//运行消息过滤
		go reg.filter()
	}

}

//通道中取出数据，进行数据过滤
func (r *reTask) filter() {

	for {
		select {
		case data := <-r.tail:
			c := r.re.FindString(data)
			if len(c) > 8 || c == "panic" {
				lg.Log.Info("开始报警,抓取到err错误数据:", data)
				//钉钉消息发送
				dingding.DDSend("日志报警！！！！日志报警！！！！" + data)
			}
		case data := <-r.gops:

			//数据写入es
			es.EsAdd(data)
			err := json.Unmarshal([]byte(data), &js)
			if err != nil {
				lg.Log.Error("gops解析成json失败", err)
			}
			if js.Cpu > 0.2 {
				//钉钉消息发送
				dingding.DDSend("服务报警！！！！服务报警！！！！" + data)
				fmt.Println("gops报警·····························································")
			}
		}
	}

}

// StartConsumer 消费者
func (r *reTask) StartConsumer(topic string, number int) {
	cfg := nsq.NewConfig()
	//cfg.LookupdPollInterval = time.Second //设置重连时间
	consumer, err := nsq.NewConsumer(topic, "01", cfg)
	if err != nil {
		lg.Log.Error("消费出错", err)
		return
	}

	// 设置消息处理函数
	consumer.AddHandler(nsq.HandlerFunc(func(message *nsq.Message) error {
		lg.Log.Info("nsq数据消费成功:", string(message.Body), "消费协程号:", number)
		//消费的数据给通道
		switch topic {
		case "gops": //走gops通道
			r.gops <- string(message.Body)
		case "cooting": //走服务日志通道
			r.tail <- string(message.Body)
		}

		return nil
	}))

	// 连接到单例nsqdr
	if err := consumer.ConnectToNSQD(r.addr); err != nil {
		lg.Log.Info(err)
	}

	<-consumer.StopChan
	return
}
